package jp.aist.gtrc.plus.scheduler.job;

import org.apache.log4j.Logger;

import java.util.Arrays;
import java.util.Collection;
import java.util.Map;

import jp.aist.gtrc.plus.scheduler.SchedulerOptions;
import jp.aist.gtrc.plus.scheduler.status.JobStatus;
import jp.aist.gtrc.plus.scheduler.sys.SchedulerException;
import jp.aist.gtrc.plus.scheduler.util.CollectionUtil;
import jp.aist.gtrc.plus.scheduler.util.Sorter;
import jp.aist.gtrc.plus.scheduler.util.SortKey;


public class JobSorter {

        protected static final Logger logger = Logger.getLogger(JobSorter.class);

	private final Sorter<JobStatus> sorter;

	private static final String TITLE_JOB_SORTKEY = "JobSortKey";
	private static final String DELIM_KEYS = ",\\s*";

	public JobSorter() {

		sorter = new Sorter<JobStatus>();

	}

	public JobSorter(SchedulerOptions options) throws SchedulerException {
                this();

                assert(options != null);
		Map<String, String> confs = options.getConfig();
		if (confs != null) {
			setJobSorter(confs);
		} else {
			// config file doesn't exist.
			// We sort jobs same as default policy of Torque's fifo scheduler.
			logger.info("config file not specified, sort jobs as Torque does.");
			addJobSortKey(JobSortKey.QueuePriority);
			addJobSortKey(JobSortKey.ByQueue);
			addJobSortKey(JobSortKey.JobPriority);
		}

		/*
		 * If there are free(idle) execute nodes, jobs are started
		 * as soon as submitted regardless sort orders specified above.
		 *
		 * If all nodes are busy at submit, submitted jobs are queued.
		 * After some running job is exited, regardless order in queue,
		 * a job with highest priority in queue is started.
		 *
		 * start		owner's			job's requested
		 * order	group	name	priority	time
		 * ------------------------------------------
		 * 1		prof.	knuth	 0			60min
		 * 2		doct.	nakada	 100		10min
		 * 3		doct.	nakada	 0			60min
		 * 4		doct.	takefusa 500		5min
		 * 5		users	someoneX 1000		15min
		 * 6		users	someoneY 0			15min
		 * ....
		 *
		 * Job owner's group is a primary group of owner
		 * which specified in /etc/passwd & /etc/group.
		 *
		 * Job priority is specified by 'qsub -p n'.
		 * Larger n	means higher priority (0 is default, 1023 is highest).
		 *
		 * Professor knuth is *not* in OwnersName list above,
		 * but his job is started at first because he is in professors group
		 * which has highest priority and owner's group is primary sort key.
		 * Both nakada & takefusa are in same doctors group,
		 * but nakada has higher priority than takefusa.
		 * nakada's jobs are started in job priority order.
		 * someoneX in users group requested highest (=1000) priority,
		 * but it runs after professors&doctors jobs are started.
		 */

	}

	public void addJobSortKey(JobSortKey key) {

		sorter.addSortKey(key);
		logger.info("Added JobSortKey: " + key);

	}

        public void addJobSortKey(SortKey<JobStatus> key) {
                sorter.addSortKey(key);
		logger.info("Added SortKey: " + key);
        }
        

	private void setJobSorter(Map<String, String> confs) {

		assert(confs != null);

		String jobKeyConf = confs.get(TITLE_JOB_SORTKEY);
		if (jobKeyConf == null)
			return;

		for (String keyName : jobKeyConf.split(DELIM_KEYS)) {
			JobSortKey key = JobSortKey.valueOf(keyName);
			if (key == null) {
				logger.warn("Unknown job sort key: "
						+ keyName + ", ignored");
				continue;
			}
			if (sorter.containsKey(key)) {
				logger.warn("Duplicate job sort key: "
						+ keyName + ", refers first one");
				continue;
			}

			if (key.needOrders()) {
				/*
				 * Set Job Owner's (group/host/domain) Orders
				 *	If keyName is "OwnersName", key.getConfigTitle() is
				 *	"OwnersOrder", and so on.
				 */
				String names = confs.get(key.getConfigTitle());
				if (names == null){
					logger.warn(keyName + " needs to specify"
							+ key.getConfigTitle());
					continue;
				}
				String[] orders = names.split(DELIM_KEYS);
				if (orders.length == 0) {
					logger.warn(key.getConfigTitle()
							+ " needs to specify order names");
					continue;
				}
				key.setOrders(orders);
			}

			// use addFirst() to sort by *least* important key at *first*,
			// sort by *most* important key at *last*.
			// See sortJobs()
			addJobSortKey(key);
		}
		if (sorter.size() == 0){
			logger.warn("No valid job sort key. Run jobs as Fifo.");
		}

	}

        public Collection<JobStatus> sort(Collection<JobStatus> jobs) {

                if ( (CollectionUtil.isEmpty(jobs)) || (sorter.size() == 0) )
                        return jobs;

                JobStatus[] jobArray = jobs.toArray(new JobStatus[jobs.size()]);
                sorter.sort(jobArray);
                return Arrays.asList(jobArray);
        }



}
